package org.midy.infrastructure;

import org.midy.infrastructure.tools.Logger;

import java.io.Closeable;
import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 来用任务关联和协调：
 * 1. 任务分配的负载均衡
 * 2. 资源同意管理(初始化、关闭)
 */
public class IOEventDrivenContainer implements EventDrivenTaskAllocation, Closeable {
    private final AtomicInteger INDEX = new AtomicInteger(0);
    private final AtomicInteger status = new AtomicInteger(0);
    private final List<IOEventDrivenWorker> workerList = new ArrayList<>();
    /**
     * status: 原子类型，状态不可逆的状态机来保证整个生命周期的合理性
     * 0: 待初始化阶段
     * 1：初始化完成
     * 2：容器进入关闭阶段
     * 3：容器已经彻底关闭，生命周期已结束
     */
    private final Integer workerNum;

    private final String name;

    private final EventDrivenTaskAllocation eventDrivenTaskAllocation;

    /**
     * 该接口是用来处理
     */
    private final InformationDistribution informationDistribution;

    public IOEventDrivenContainer(String name, Integer workerNum, InformationDistribution informationDistribution, EventDrivenTaskAllocation eventDrivenTaskAllocation) {
        if (Objects.nonNull(name)) {
            this.name = name;
        } else {
            this.name = "eventDrivenContainer-[" + INDEX.incrementAndGet() + "]";
        }

        this.workerNum = workerNum;
        this.informationDistribution = Objects.isNull(informationDistribution) ? new DefaultInformationDistribution() : informationDistribution;
        this.eventDrivenTaskAllocation = Objects.nonNull(eventDrivenTaskAllocation) ? eventDrivenTaskAllocation : this;
        initWorkerGroup(workerNum); // 初始化工作组
    }

    /**
     * IO工作者将收到的IO信息封装成任务通过阻塞队列进行派发
     *
     * @param workerNum 初始化N个线程来监听Selector并处理IO事件
     */
    public synchronized void initWorkerGroup(int workerNum) {
        int sv = status.get();

        if (sv != 0) {
            throw new RuntimeException("当前工作组容器不能重复初始化！");
        }

        if (workerNum <= 0) {
            throw new IllegalArgumentException(" worker number 不能小于1");
        }

        for (int index = 0; index < workerNum; index++) {
            IOEventDrivenWorker newIOEventDrivenWorker = new IOEventDrivenWorker(name + "-[" + index + "]", informationDistribution, this.eventDrivenTaskAllocation);
            workerList.add(newIOEventDrivenWorker);
        }

        if (!status.compareAndSet(0, 1)) { // 因为加锁了，所以不会出现
            throw new IllegalArgumentException("初始化并发异常");
        }
    }

    private final AtomicInteger index = new AtomicInteger(0);

    /**
     * 实现任务的[轮训]负载均衡分配：可以实现：
     * 1. 基于可分配资源数来轮训、权重轮训
     * 2. 随机数分配、支持权重随机数
     * 3. 根据每个资源的负载情况来适当分配
     *
     * @param socketChannel 需要被监听的网络连接
     * @param selectionKey  待监听的网络连接事件
     * @return boolean
     * @throws RuntimeException 其他：可以扩展一个接口用来实现负载均衡策略
     */
    @Override
    public boolean allocation(SelectableChannel socketChannel, int selectionKey) throws RuntimeException {
        if (isClosed()) {
            return false;
        }

        int size = workerList.size();
        int idx = 0;
        if (size != 1) {
            int order;
            do {
                idx = index.get();
                order = idx + 1;
                if (order == workerList.size()) {
                    order = 0;
                }
            } while (!index.compareAndSet(idx, order));
        }

        IOEventDrivenWorker ioEventDrivenWorker = workerList.get(idx);
        return ioEventDrivenWorker.register(socketChannel, selectionKey);
    }

    @Override
    public boolean allocation(SelectableChannel socketChannel, int selectionKey, byte[] writeData) throws RuntimeException {
        if (isClosed()) {
            return false;
        }

        int size = workerList.size();
        int idx = 0;
        if (size != 1) {
            int order;
            do {
                idx = index.get();
                order = idx + 1;
                if (order == workerList.size()) {
                    order = 0;
                }
            } while (!index.compareAndSet(idx, order));
        }

        IOEventDrivenWorker ioEventDrivenWorker = workerList.get(idx);
        return ioEventDrivenWorker.register(socketChannel, null, selectionKey, writeData);
    }

    /**
     * 关闭容器：
     * 1. 支持并发安全处理
     * 2. 支持关闭所有Selector
     * 3. 支持关闭所有的SocketChannel
     *
     * @throws IOException 原关机接口支持抛出关闭资源失败
     */
    @Override
    public void close() throws IOException {
        boolean isClosed = false;
        while (!isClosed) {
            int expectValue = status.get();
            isClosed = expectValue > 1;

            if (isClosed) {
                break;
            }

            if (status.compareAndSet(expectValue, 2)) {/* 执行一些关闭操作 */
                try {
                    for (IOEventDrivenWorker worker : workerList) {
                        try {
                            worker.close();
                        } catch (Throwable t) {
                            Logger.warn("协作关闭线程失败: %s", t.getMessage());
                        }
                    }
                } finally {
                    isClosed = true;
                }
                break;
            }
        }
    }

    /**
     * 用来判定当前容器状态
     *
     * @return 容器的状态
     */
    public boolean isClosed() {
        return status.get() > 1;
    }
}
